package cn.texous.demo.dmqs.executor;

import cn.texous.demo.dmqs.producer.KafkaProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 异步线程执行器
 *
 * @author leven
 * @since
 */
@Component
public class AsyncTaskExecutor {
    private static Logger logger = LoggerFactory.getLogger(AsyncTaskExecutor.class);

    private ExecutorService executorService;

    private static final int WORK_QUEUE_SIZE = 0x4E2000;
    private static final int CORE_POOL_SIZE = 5;
    private static final int MAXIMUM_POOL_SIZE = 5;
    private static final long KEEP_ALIVE_TIME = 0L;


    @Autowired
    private KafkaProducer kafkaProducer;

    @PostConstruct
    public void init() {
        executorService = new ThreadPoolExecutor(CORE_POOL_SIZE,
                MAXIMUM_POOL_SIZE, KEEP_ALIVE_TIME, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>(WORK_QUEUE_SIZE), new MyThreadFactory());
        logger.info("succ to init asyncTaskExecutor ! ! !");
    }

    /***/
    @PreDestroy
    public void destory() {
        if (executorService != null) {
            executorService.shutdown();
        }
    }

    /**
     * 发送消息到kafka
     *
     * @param msg 发送到kafka中的消息参数集合
     */
    public void sendRequsetMsg(String topic, String key, Object msg) {
        executorService.execute(new KafkaMsgTask(topic, key, msg));
    }

    private class KafkaMsgTask implements Runnable {

        private String topic;
        private String key;
        private Object msg;

        public KafkaMsgTask(String topic, String key, Object msg) {
            this.topic = topic;
            this.key = key;
            this.msg = msg;
        }

        @Override
        public void run() {
            kafkaProducer.send(topic, key, msg);
        }
    }

    private static class MyThreadFactory implements ThreadFactory {

        private AtomicInteger ids = new AtomicInteger(0);

        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r);
            thread.setDaemon(true);
            thread.setName("async-task-executor-" + ids.getAndIncrement());
            return thread;
        }

    }
}
